package com.fanxuankai.kafka.connect.sink.redis.consumer;

import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.sink.SinkRecord;
import org.springframework.data.redis.core.RedisTemplate;

import java.util.*;

/**
 * @author fanxuankai
 */
public class SpringSinkRecordConsumer implements SinkRecordConsumer {
    private final RedisTemplate<String, Object> redisTemplate;

    public SpringSinkRecordConsumer(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void accept(Collection<SinkRecord> sinkRecords) {
        for (SinkRecord sinkRecord : sinkRecords) {
            if (sinkRecord.value() == null) {
                redisTemplate.opsForSet()
                        .remove(sinkRecord.topic(), sinkRecord.key()
                                .toString());
                redisTemplate.delete(sinkRecord.topic() + ":" + sinkRecord.key()
                        .toString());
            } else {
                Schema valueSchema = sinkRecord.valueSchema();
                Struct value = (Struct) sinkRecord.value();
                List<Field> fields = valueSchema.fields();
                Map<String, String> map = new HashMap<>(fields.size());
                for (Field field : fields) {
                    Object fieldValue = value.get(field);
                    String hv;
                    if (fieldValue == null) {
                        hv = null;
                    } else if (fieldValue instanceof Date) {
                        hv = String.valueOf(((Date) fieldValue).getTime());
                    } else {
                        hv = fieldValue.toString();
                    }
                    map.put(field.name(), hv);
                }
                redisTemplate.opsForHash()
                        .putAll(sinkRecord.topic() + ":" + sinkRecord.key()
                                .toString(), map);
                redisTemplate.opsForSet()
                        .add(sinkRecord.topic(), sinkRecord.key()
                                .toString());
            }
        }
    }
}
